#!/bin/bash
# File Name: diff_count.sh
# Author: moshan
# mail: mo_shan@yeah.net
# Created Time: 2019-02-22 16:46:26
# Function: 对比master与slave的数据行是否一致
#########################################################################
:<<comment
变量注释:
mysql_user    : 连接mysql用户名;
mysql_passwd  : 连接mysql的密码;
mysql_port1   : 第一个节点的端口号;
mysql_port2   : 第二个节点的端口号;
log_dir       : 日志文件存放的目录;
log_partition : 日志文件存放的目录所在的分区
log_par_size  : 日志文件预计的大小, 如果该分区所剩的空间大小小于该值就退出, 默认是10, 表示10GB
mysql_host1   : 第一个节点的ip地址
mysql_host2   : 第二个节点的ip地址
count         : check数据行的最小单位，默认是10000行
check_no_pk   : 该参数控制是否check没有主键的表, 1检查, 0不检查
注意：默认以第一个节点为参照，即程序会先获取到第一个节点的业务表
comment
mysql_user="test"
mysql_passwd="test"
mysql_port1="3306"
mysql_port2="3311"
mysql_host1="172.28.84.239"
mysql_host2="172.28.84.239"
count=1000000
log_dir="/tmp"
log_partition="/dev/mapper/mint--vg-root"
log_par_size="10"
check_no_pk=1
mysql_path="/data/mysql/base/bin/mysql"
db="${1}"

#以下无需编辑修改
tmp_file="${log_dir}/tmp_${db}.log"
error_log_file="${log_dir}/error_result.log"
error_log_table_dir="${log_dir}/table"
[ ! -d "${error_log_table_dir}" ] && mkdir -p ${error_log_table_dir}
info_log_file="${log_dir}/info_result.log"
mysql_port="${mysql_port1}"
host_list=(${mysql_host1} ${mysql_host2})
mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_passwd} -h${host_list[0]} -P${mysql_port1} ${db_name}"
${mysql_comm} -NBe "select 1;" >/dev/null 2>&1
if [ $? -ne 0 ]
then
        echo
        echo -e "\033[31m[$(date "+%F %H:%M:%S")] [ERROR] Can't connect to MySQL server on '${mysql_user}@${mysql_host1}' for mysql_port:${mysql_port}\033[0m"
        echo
        exit 1
else
        >${error_log_file}
fi

function f_check_partition()
{
	free_space="$(df|grep ${log_partition} 2> /dev/null|awk -v size="${log_par_size}" '{if ($4/1024/1024>size){print 1}else print 0}')"
	if [ "${free_space}x" == "0x" ]
	then
		echo -e "\033[33m[$(date "+%F %H:%M:%S")] [WARN] 该日志目录所在分区[${log_partition}]所剩空间少于${log_par_size}GB, 程序将退出运行\033[0m" |tee -a ${error_log_file}
		exit
	fi
}

function f_check_diff()
{
	db_name="${1}"
	mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_passwd} -h${host_list[0]} -P${mysql_port1} ${db_name} -A"
	table=($(${mysql_comm} -NBe "show tables" 2>/dev/null))
	for ((i=0;i<${#table[@]};i++))
	do
		mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_passwd} -h${host_list[0]} -P${mysql_port1} ${db_name} -A"
		pri_name="$(${mysql_comm} -NBe "desc ${table[${i}]}" 2>/dev/null|awk 'BEGIN{str2="PRI";pri=0;}{str=$4;if (str==str2){pri+=1;pri_str=(pri_str"+"$1);pri_type=(pri_type"+"$2)}}END{if(pri>0){print pri_str":"pri_type":"pri}else{print "NULL"}}')"
		pri_type=($(awk -F: '{print $2}' <<< "${pri_name}"|tr "+" " "))
		pri_count="$(awk -F: '{print $3}' <<< "${pri_name}")"
		pri_name=($(awk -F: '{print $1}' <<< "${pri_name}"|tr "+" " "))
		select_str=""
		[ "${pri_name}x" == "NULLx" ] && pri_count=0
		for ((tmp=0;tmp<${pri_count};tmp++))
		do
			if [ "$(grep -c "int" <<< "${pri_type[${tmp}]}")x" != "1x" ]
			then
				select_str="${select_str},concat(\"'\",${pri_name[${tmp}]},\"'\")"
			else
				select_str="${select_str},${pri_name[${tmp}]}"
			fi
		done
		select_str="$(sed 's/^,//g' <<< "${select_str}")"
		if [ "${pri_name}x" != "NULLx" -o "${check_no_pk}x" == "1x" ]
		then
			pri_name="($(echo "${pri_name[@]}"|tr " " ","))"
			> ${tmp_file}
			if [ "${pri_count}x" != "0x" ]
			then
				sql_str_tmp="select ${select_str} from ${table[${i}]};"
				echo ${sql_str_tmp}|${mysql_comm} -NB > ${tmp_file} 2>/dev/null
				[ $? -ne 0 ] && echo -e "\033[31m[$(date "+%F %H:%M:%S")] [INFO] SQL语句语法错误：${sql_tmp_tmp}, 跳过该表的检查\033[0m" && continue
			fi
			for ((k=0;k<=$(wc -l < ${tmp_file});k=k+${count}))
			do
				[ "${k}x" == "0x" ] && k=1
				if [ "${pri_count}x" != "0x" ]
				then
					where_str="$(sed -n "${k},$((${k}+${count}-1))p" ${tmp_file}|tr "\n" ","|sed 's/,/),(/g'|sed 's/^/(/g;s/,($//g'|tr "\t" ","|awk -v pri_str="${pri_name}" '{print "where "pri_str" in ("$0");"}')"
				else
					where_str=""
					echo -e "\033[33m[$(date "+%F %H:%M:%S")] [WARN] ${db_name}.${table[${i}]}:没有主键。\033[0m" |tee -a ${info_log_file}
				fi
				col_str="$(${mysql_comm} -NBe "desc ${table[${i}]};" 2>/dev/null|awk '{print $1}'|tr "\n" ",")"
				col_str_isnull="$(${mysql_comm} -NBe "desc ${table[${i}]};" 2>/dev/null|awk '{print "ISNULL("$1")"}'|tr "\n" ","|sed 's#\(.*\),$#\1#g')"
				for ((j=0;j<${#host_list[@]};j++))
				do
				{
						if [ "${j}x" == "0x" ]
						then
								mysql_port="${mysql_port1}"
						else
								mysql_port="${mysql_port2}"
						fi
						mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_passwd} -h${host_list[${j}]} -P${mysql_port} ${db_name} -A"
						echo "set tx_isolation='REPEATABLE-READ';SELECT '${table[${i}]}', COUNT(*) AS cnt, COALESCE(LOWER(CONV(BIT_XOR(CAST(CRC32(CONCAT_WS('#',${col_str}CONCAT(${col_str_isnull}))) AS UNSIGNED)), 10, 16)), 0) AS crc FROM ${db_name}.${table[${i}]} ${where_str};"|${mysql_comm} -NB 2>/dev/null|awk '{print $1":"$2":"$3}' > ${host_list[${j}]}_${mysql_port}_${db_name}.log
				} &
				done
				wait
				if [ $(wc -l < ${tmp_file}) -eq 0 ]
				then
					start_count=0
				else
					start_count=${k}
				fi
				if [ $((${k}+${count}-1)) -ge $(wc -l < ${tmp_file}) ]
				then
					end_count=$(wc -l < ${tmp_file})
				else
					end_count=$((${k}+${count}-1))
				fi
				if [ "$(md5sum ${host_list[0]}_${mysql_port1}_${db_name}.log|awk '{print $1}')x" == "$(md5sum ${host_list[1]}_${mysql_port2}_${db_name}.log|awk '{print $1}')x" ]
				then
					echo -e "\033[32m[$(date "+%F %H:%M:%S")] [OK] ${db_name}.${table[${i}]}[${start_count},${end_count}]:数据一致。\033[0m"
				else
					echo -e "\033[31m[$(date "+%F %H:%M:%S")] [ERROR] ${db_name}.${table[${i}]}[${start_count},${end_count}]:数据不一致。第一个节点的数据行是：$(awk -F: '{print $2}' ${host_list[0]}_${mysql_port1}_${db_name}.log)，第二个节点的数据行是：$(awk -F: 'BEGIN{sum=$2}END{if (sum==""){print "0"}else {print $2}}' ${host_list[1]}_${mysql_port2}_${db_name}.log)。\033[0m" |tee -a ${error_log_file}
					echo -e "\033[33m[$(date "+%F %H:%M:%S")] [SQL] set tx_isolation='REPEATABLE-READ';SELECT '${table[${i}]}', COUNT(*) AS cnt, COALESCE(LOWER(CONV(BIT_XOR(CAST(CRC32(CONCAT_WS('#',${col_str}CONCAT(${col_str_isnull}))) AS UNSIGNED)), 10, 16)), 0) AS crc FROM ${db_name}.${table[${i}]} ${where_str};" >> ${error_log_table_dir}/${host_list[0]}_${mysql_port1}_${db_name}_${table[${i}]}.log
				fi
				[ -f "${host_list[0]}_${mysql_port1}_${db_name}.log" ] && rm -f ${host_list[0]}_${mysql_port1}_${db_name}.log
				[ -f "${host_list[1]}_${mysql_port2}_${db_name}.log" ] && rm -f ${host_list[1]}_${mysql_port2}_${db_name}.log
			done
		else
			echo -e "\033[33m[$(date "+%F %H:%M:%S")] [WARN] ${db_name}.${table[${i}]}:没有主键。\033[0m" |tee -a ${error_log_file}
		fi
	done
}

f_check_partition
f_check_diff ${1}
